package topology;

import kafka.newstyle.Producer;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.storm.trident.operation.BaseFunction;
import org.apache.storm.trident.operation.TridentCollector;
import org.apache.storm.trident.operation.TridentOperationContext;
import org.apache.storm.trident.tuple.TridentTuple;
import org.apache.storm.tuple.Values;

import java.io.*;
import java.nio.charset.StandardCharsets;
import java.util.List;
import java.util.Map;

public class IpAreaIspBolt extends BaseFunction {

    List<Long> minList ;
    List<Long> maxList;
    List<String> guojiaList;
    List<String> shengfenList;
    List<String> chengshiList;

    @Override
    public void prepare(Map<String, Object> conf, TridentOperationContext context) {
        // 读取地址信息，初始化minList,maxList,addressList变量
        String filePath = Producer.class.getClassLoader().getResource("data/ip_area_isp.txt").getFile();
        File ip_area_isp = new File(filePath);
        InputStreamReader isr = null;
        try {
            isr = new InputStreamReader(new FileInputStream(ip_area_isp), StandardCharsets.UTF_8);
        } catch (FileNotFoundException e) {
            e.printStackTrace();
        }
        BufferedReader br = new BufferedReader(isr);
        String lineTxt;
        try {
            while ((lineTxt = br.readLine()) != null) {
                String[] diqu=lineTxt.split("\t");
                for(int i=0;i<diqu.length;i++) {
                    minList.add(Long.valueOf(diqu[4]));
                    maxList.add(Long.valueOf(diqu[5]));
                    shengfenList.add(diqu[1]);
                    chengshiList.add(diqu[2]);
                    guojiaList.add(diqu[0]);
                }
            }
        } catch (FileNotFoundException e) {
            e.printStackTrace();
        } catch (IOException e) {
            e.printStackTrace();
        }
    }

    @Override
    public void execute(TridentTuple tuple, TridentCollector collector) {
            int ip=tuple.getInteger(0);
            for (int i=0;i< minList.size();i++)
            if (ip>=minList.get(i) && ip<= maxList.get(i)){
                collector.emit(new Values(guojiaList.get(i),shengfenList.get(i),chengshiList.get(i)));
            }





    }
}
